package net.xo68.destiny.core.timingwheel.impl;

import lombok.extern.slf4j.Slf4j;
import net.xo68.destiny.contract.common.SystemConstants;
import net.xo68.destiny.core.timingwheel.*;
import net.xo68.destiny.core.timingwheel.exception.TimingWheelShutdownException;
import net.xo68.destiny.core.timingwheel.handle.TimeoutEventHandler;
import net.xo68.destiny.core.util.LocalDateTimeUtils;
import net.xo68.destiny.core.util.ThreadUtils;

import java.io.Serializable;
import java.util.Collections;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 时光轮调度抽象类
 * @author wuxie
 * @version 2020-10-19
 */
@Slf4j
public abstract class AbstractTimingWheel<T extends Serializable> implements TimingWheel<T> {

    /**
     * 初始化状态
     */
    public static final int WORKER_STATE_INIT      = 0;
    /**
     * 已启动
     */
    public static final int WORKER_STATE_STARTED   = 1;
    /**
     * 已关闭
     */
    public static final int WORKER_STATE_SHUTDOWN  = 2;

    /**
     * 状态更新者
     */
    public static final AtomicIntegerFieldUpdater<AbstractTimingWheel> workerStateUpdater     = AtomicIntegerFieldUpdater
            .newUpdater(
                    AbstractTimingWheel.class,
                    "workerState");
    /**
     * 时光轮工作状态
     */
    private volatile int workerState;

    /**
     * 超时项队列,新任务先进行此队列
     */
    private final Queue<Timeout<T>> timeouts= new ConcurrentLinkedQueue<>();
    /**
     * 准备取消的超时项队列
     */
    private final Queue<Timeout<T>> cancelledTimeouts= new ConcurrentLinkedQueue<>();

    /**
     * 即将超时项数量
     */
    private final AtomicLong pendingTimeouts= new AtomicLong(0);
    /**
     * 即将超时项的最大数量
     */
    private final long maxPendingTimeouts;

    /**
     * 时光轮启动时间,单位：毫秒数
     */
    protected volatile long startTime;
    /**
     *轮子的格子数量
     */
    private final long ticksPerWheel;
    /**
     * 轮子每个格了的间隔时间，单位毫秒
     */
    private final long tickDuration;

    /**
     * 任务序列号生成者
     */
    private final TimingWheelTaskSequencer<T> sequencer;
    /**
     * 时光轮最大下标
     */
    private final long mask;
    /**
     * 时光轮走过的格子总数
     */
    private final AtomicLong tick=new AtomicLong(0);

    /**
     * 时光轮自动旋转调度线程池服务
     */
    private final ScheduledExecutorService rotateScheduledService;
    /**
     * 时光轮容器
     */
    private final ConcurrentHashMap<Long, TimingWheelBucket<T>> wheel;

    /**
     * 任务缓冲
     */
    private final HashedTimeoutBuffer<T> hashedBuffer=new HashedTimeoutBuffer<>();

    /**
     * 时光轮任务执行器
     */
    private final TimingWheelTaskExecutor taskExecutor;
    /**
     * 超时项任务处理器
     */
    private final TimeoutEventHandler<T> eventHandler;
    /**
     * 时光轮的工作者
     */
    private final TimingWheelWorker<T> worker;
    /**
     * 压入任务的锁
     */
    private ReentrantLock pushLock=new ReentrantLock();

    public AbstractTimingWheel(String name, long ticksPerWheel, long tickDuration, TimingWheelTaskSequencer<T> sequencer, TimingWheelTaskExecutor taskExecutor, long maxPendingTimeouts,
                               TimeoutEventHandler<T> eventHandler) {
        this.ticksPerWheel = normalizeTicksPerWheel(ticksPerWheel);
        this.tickDuration=normalizeTickDuration(tickDuration);
        log.debug("ticksPerWheel:{},tickDuration:{}",this.ticksPerWheel,this.tickDuration);
        this.sequencer=sequencer;
        this.rotateScheduledService= ThreadUtils.newSingleThreadScheduled(name+"-AutoRotate");
        this.taskExecutor=taskExecutor;
        this.wheel=new ConcurrentHashMap<>();
        initWheel();
        this.mask=this.ticksPerWheel - 1;
        log.info("mask:{}",this.mask);
        this.maxPendingTimeouts=maxPendingTimeouts;
        this.eventHandler=eventHandler;
        this.worker=new TimingWheelWorker<>(this);
    }
    public AbstractTimingWheel(String name, long ticksPerWheel, long tickDuration, TimingWheelTaskSequencer<T> sequencer, int threadCount, long maxPendingTimeouts, TimeoutEventHandler<T> eventHandler) {
        this(name,ticksPerWheel,tickDuration,sequencer,new DefaultTimingWheelTaskExecutor(name, threadCount),maxPendingTimeouts,eventHandler);
    }
    public AbstractTimingWheel(String name, long ticksPerWheel, long tickDuration, TimingWheelTaskSequencer<T> sequencer, TimeoutEventHandler<T> eventHandler) {
        this(name,ticksPerWheel,tickDuration,sequencer,0,-1,eventHandler);
    }
    public AbstractTimingWheel(String name, long ticksPerWheel, TimingWheelTaskSequencer<T> sequencer, TimeoutEventHandler<T> eventHandler) {
        this(name,ticksPerWheel,0L,sequencer,eventHandler);
    }
    public AbstractTimingWheel(String name, TimingWheelTaskSequencer<T> sequencer, TimeoutEventHandler<T> eventHandler) {
        this(name,8,sequencer,eventHandler);
    }

    /**
     * 初始轮容器
     */
    private void initWheel(){
        for (long i=0L;i<this.ticksPerWheel;i++){
            this.wheel.put(i,new TimingWheelBucket<T>(i));
        }
    }


    /**
     * 将超时项推入时光轮缓存队列
     * @param triggerParameter
     * @param delay
     * @param unit
     * @return
     */
    @Override
    public Timeout<T> newTimeout(T triggerParameter, long delay, TimeUnit unit){
        //log.debug("接收超时任务： {}",triggerParameter);
        //判断时光轮工作状态，并处理
        if(getWorkerState()!=WORKER_STATE_STARTED){
            throw new TimingWheelShutdownException();
        }
        if (triggerParameter == null) {
            throw new NullPointerException("triggerParameter");
        }
        if (unit == null) {
            throw new NullPointerException("unit");
        }
        if(delay<0L){
            delay=0L;
        }

        pendingTimeouts.incrementAndGet();
        if (maxPendingTimeouts > 0 && this.pendingTimeouts.get() > maxPendingTimeouts) {
            pendingTimeouts.decrementAndGet();
            throw new RejectedExecutionException("Number of pending timeouts (" + pendingTimeouts
                    + ") is greater than or equal to maximum allowed pending "
                    + "timeouts (" + maxPendingTimeouts + ")");
        }
        //根据当前时间修改最终延迟时间线
        long deadline=System.currentTimeMillis()+unit.toMillis(delay)-this.startTime;
        // Guard against overflow.
        if (delay > 0L && deadline < 0L) {
            deadline = 0L;
        }
        String sequence=this.sequencer.sequence(triggerParameter);
        TimingWheelTimeout<T> timeout = new TimingWheelTimeout<>(sequence, this, unit.toMillis(delay), deadline, triggerParameter);

        //压入缓冲
        this.hashedBuffer.put(sequence, timeout);
        this.timeouts.add(timeout);
        return timeout;
    }

    @Override
    public boolean cancelTimeout(String taskSequence){
        if(this.hashedBuffer.containsKey(taskSequence)){
            this.cancelledTimeouts.add(this.hashedBuffer.get(taskSequence));
            return true;
        }
        return false;
    }

    @Override
    public int getWorkerState() {
        return workerState;
    }

    @Override
    public long getStartTime() {
        return startTime;
    }

    @Override
    public long getTicksPerWheel() {
        return ticksPerWheel;
    }

    @Override
    public TimingWheelTaskSequencer<T> getSequencer() {
        return sequencer;
    }

    @Override
    public long getTickDuration() {
        return tickDuration;
    }

    @Override
    public long getMask() {
        return mask;
    }

    @Override
    public AtomicLong getTick() {
        return tick;
    }

    /**
     * 跳到下一格子(刻度)
     * @return
     */
    @Override
    public long nextTick(){
        tick.incrementAndGet();
        return this.tick.get();
    }



    @Override
    public ConcurrentHashMap<Long, TimingWheelBucket<T>> getWheel() {
        return wheel;
    }

    @Override
    public HashedTimeoutBuffer<T> getHashedBuffer() {
        return hashedBuffer;
    }

    @Override
    public TimingWheelTaskExecutor getTaskExecutor() {
        return taskExecutor;
    }

    @Override
    public boolean isRunning(){
        return this.workerState==WORKER_STATE_STARTED;
    }

    @Override
    public Queue<Timeout<T>> getTimeouts() {
        return timeouts;
    }

    @Override
    public Queue<Timeout<T>> getCancelledTimeouts() {
        return cancelledTimeouts;
    }

    @Override
    public AtomicLong getPendingTimeouts() {
        return pendingTimeouts;
    }

    /**
     * 超时任务处理器
     * @return
     */
    @Override
    public TimeoutEventHandler<T> getEventHandler() {
        return eventHandler;
    }

    @Override
    public void start(){
        switch (workerStateUpdater.get(this)) {
            case WORKER_STATE_INIT:
                if (workerStateUpdater.compareAndSet(this, WORKER_STATE_INIT, WORKER_STATE_STARTED)) {
                    //启动调度 worker
                    final long currentTime=System.currentTimeMillis();
                    final long delay=1000 - currentTime % 1000 + SystemConstants.INITIAL_DELAY_SECOND*1000 - 1000;
                    this.startTime=currentTime+delay;
                    log.info("时光轮worker准备启动：{}，将于{}毫秒后启动，准时启动时间:{}",LocalDateTimeUtils.parseDateTime(currentTime),delay,LocalDateTimeUtils.parseDateTime(this.startTime));
                    this.rotateScheduledService.scheduleAtFixedRate(this.worker,delay,this.tickDuration,TimeUnit.MILLISECONDS);
                }
                break;
            case WORKER_STATE_STARTED:
                break;
            case WORKER_STATE_SHUTDOWN:
                throw new IllegalStateException("cannot be started once stopped");
            default:
                throw new Error("Invalid WorkerState");
        }
    }

    @Override
    public Set<Timeout<T>> shutdown(){
        if (!workerStateUpdater.compareAndSet(this, WORKER_STATE_STARTED, WORKER_STATE_SHUTDOWN)){
            return Collections.emptySet();
        }
        try {
            this.worker.getShutdowned().await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        Set<Timeout<T>> unProcessTimeouts=shutdownRotateScheduled();

//        for (Timeout<T> timeout: unProcessTimeouts){
//            this.hashedBuffer.remove(timeout.taskSequence());
//        }
        shutdownTaskExecutor();
        return unProcessTimeouts;
    }


    /**
     * 关闭自旋调度线程池服务
     */
    private  Set<Timeout<T>> shutdownRotateScheduled() {
        //关闭自旋调度线程池服务
        this.rotateScheduledService.shutdown();
        return this.worker.getUnprocessedTimeouts();
    }

    /**
     * 关闭超时项执行线程池服务
     */
    private void shutdownTaskExecutor() {
        //关闭自旋调度线程池服务
        this.taskExecutor.shutdown();
    }

    /**
     * 标准化轮子的格子数量 必须是2的次方
     * @param ticksPerWheel
     * @return
     */
    private static long normalizeTicksPerWheel(long ticksPerWheel) {
        if(ticksPerWheel<8L){
            ticksPerWheel=8L;
        }
        long normalizedTicksPerWheel = 1L;
        while (normalizedTicksPerWheel < ticksPerWheel) {
            normalizedTicksPerWheel <<= 1L;
        }
        return normalizedTicksPerWheel;
    }

    /**
     * 修正格子间隔时间
     * @param tickDuration
     * @return
     */
    private static long normalizeTickDuration(long tickDuration) {
        long normalizedTickDuration = tickDuration;
        if(tickDuration<=0L){
            normalizedTickDuration = 1000L;
        }else if(tickDuration < 100L) {
            normalizedTickDuration = 100L;
        }
        return normalizedTickDuration;
    }
}
